package cn.monkey.transport.netty.server.codec;

import cn.monkey.transport.core.ExchangeManager;
import cn.monkey.transport.core.server.FilterChain;
import cn.monkey.transport.netty.NettyExchange;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ExchangeChannelMessageFilterInboundHandler<EXCHANGE extends NettyExchange, PKG> extends ExchangeChannelInboundHandler<EXCHANGE, PKG> {

    private static final Logger log = LoggerFactory.getLogger(ExchangeChannelMessageFilterInboundHandler.class);
    protected final FilterChain<? super EXCHANGE, PKG> filterChain;

    public ExchangeChannelMessageFilterInboundHandler(ExchangeManager<Channel, EXCHANGE> exchangeManager,
                                                      boolean autoRelease,
                                                      FilterChain<? super EXCHANGE, PKG> filterChain) {
        super(exchangeManager, autoRelease);
        this.filterChain = filterChain;
    }

    public ExchangeChannelMessageFilterInboundHandler(ExchangeManager<Channel, EXCHANGE> exchangeManager,
                                                      FilterChain<? super EXCHANGE, PKG> filterChain) {
        super(exchangeManager);
        this.filterChain = filterChain;
    }


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, PKG msg) {
        EXCHANGE exchange;
        String key = this.key(ctx);
        try {
            exchange = this.exchangeManager.findOrCreate(key, ctx.channel());
            if (log.isDebugEnabled()) {
                log.debug("find or create exchange: {}", exchange.id());
            }
        } catch (Throwable e) {
            log.error("find or create exchange error, key: {}, channel: {}: \n", key, ctx.channel(), e);
            ctx.close();
            return;
        }
        this.filterChain.doFilter(exchange, msg);
    }
}
